package cn.junechiu.learnrxjava;

import android.os.Bundle;
import android.support.v7.app.AppCompatActivity;
import android.util.Log;
import android.widget.TextView;

import butterknife.BindView;
import butterknife.ButterKnife;
import cn.junechiu.learnrxjava.bean.GitUser;
import cn.junechiu.learnrxjava.net.Api;
import cn.junechiu.learnrxjava.net.RetrofitManager;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.schedulers.Schedulers;

/**
 * observable.subscribe(observer);
 * // 或者：
 * observable.subscribe(subscriber);
 * <p>
 * subscribe()有多个重载的方法
 * public final Disposable subscribe() {}
 * 下游不关心任何事件,你上游尽管发你的数据去吧
 * public final Disposable subscribe(Consumer<? super T> onNext) {}
 * 带有一个Consumer参数的方法表示下游只关心onNext事件, 其他的事件我假装没看见
 * public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError) {}
 * public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete) {}
 * public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete, Consumer<? super Disposable> onSubscribe) {}
 * public final void subscribe(Observer<? super T> observer) {}
 * <p>
 * Schedulers.io() 代表io操作的线程, 通常用于网络,读写文件等io密集型的操作
 * Schedulers.computation() 代表CPU计算密集型的操作, 例如需要大量计算的操作
 * Schedulers.newThread() 代表一个常规的新线程
 * AndroidSchedulers.mainThread() 代表Android的主线程
 */
public class MainActivity extends AppCompatActivity {

    private String TAG = getClass().getSimpleName();

    @BindView(R.id.text)
    TextView textView;

    private StringBuilder string = new StringBuilder();

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        ButterKnife.bind(this);

        sample1();
        sample2();
        sample3();
        sample4("abc20899");
    }

    public void sample1() {
        // 创建被观察者、数据源
        Observable observable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();
            }
        });

        //创建管擦者
        Observer observer = new Observer() {
            @Override
            public void onSubscribe(Disposable d) {
            }

            @Override
            public void onNext(Object value) {
                string.append(value + " ");
                Log.d("onNext", "value: " + value);
            }

            @Override
            public void onError(Throwable e) {
            }

            @Override
            public void onComplete() {
                Log.d("onComplete", "finish");
                textView.setText("onComplete: " + string);
            }
        };
        //建立连接
        observable.subscribe(observer);

        /**
         * 1.上游可以发送无限个onNext, 下游也可以接收无限个onNext.
         * 2.当上游发送了一个onComplete/onError后, 上游onComplete/onError之后的事件将会继续发送,
         * 而下游收到onComplete事件之后将不再继续接收事件.
         * 3.上游可以不发送onComplete或onError.
         * 4.onComplete和onError必须唯一并且互斥, 即不能发多个onComplete, 也不能发多个onError,
         * 也不能先发一个onComplete, 然后再发一个onError, 反之亦然
         * */
    }

    public void sample2() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.d("emitter", "1");
                emitter.onNext(1);
                Log.d("emitter", "2");
                emitter.onNext(2);
                Log.d("emitter", "3");
                emitter.onNext(3);
                Log.d("emitter", "complete");
                emitter.onComplete();
                Log.d("emitter", "4");
                emitter.onNext(4);
            }
        }).subscribe(new Observer<Integer>() {
            int i = 0;
            Disposable mDisposable;  //管道开关

            @Override
            public void onSubscribe(Disposable d) {
                Log.d("onSubscribe", "");
                mDisposable = d;
            }

            @Override
            public void onNext(Integer value) {
                Log.d("onNext", "value: " + value);
                i += 1;
                if (i == 2) {
                    mDisposable.dispose(); //切断下游 停止接收数据
                    Log.d("isDisposed", ": " + mDisposable.isDisposed());
                }
            }

            @Override
            public void onError(Throwable e) {
                Log.d("onError", ": ");
            }

            @Override
            public void onComplete() {
                Log.d("onComplete", ": ");
            }
        });
    }

    public void sample3() {
        /**
         * 正常情况下, 上游和下游是工作在同一个线程中的, 也就是说上游在哪个线程发事件, 下游就在哪个线程接收事件.
         * */
        Observable observable = Observable.create(new ObservableOnSubscribe() {
            @Override
            public void subscribe(ObservableEmitter emitter) throws Exception {
                //Observable thread is : main
                Log.d(TAG, "Observable thread is : " + Thread.currentThread().getName());
                Log.d(TAG, "emit 1");
                emitter.onNext("haha");
            }
        });

        Consumer consumer = new Consumer() {
            @Override
            public void accept(Object o) throws Exception {
                //Observer thread is :main
                Log.d(TAG, "Observer thread is :" + Thread.currentThread().getName());
                Log.d(TAG, "onNext: " + o.toString());
            }
        };

//        observable.subscribe(consumer);  // 默认在同一个线程中
        /**
         * 开启子线程
         * subscribeOn() 指定的是上游发送事件的线程, observeOn() 指定的是下游接收事件的线程.
         * 多次调用subscribeOn() 只有第一次的有效, 其余的会被忽略
         * 每调用一次observeOn() , 下游的线程就会切换一次.
         * */
//        observable.subscribeOn(Schedulers.newThread())
//                .observeOn(AndroidSchedulers.mainThread())
//                .subscribe(consumer);

        /**
         * 线程切换
         * */
        observable.subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .doOnNext(new Consumer() {
                    @Override
                    public void accept(Object o) throws Exception {
                        Log.d(TAG, "observeOn thread is :" + Thread.currentThread().getName());
                    }
                }).observeOn(Schedulers.io())
                .doOnNext(new Consumer() {
                    @Override
                    public void accept(Object o) throws Exception {
                        Log.d(TAG, "observeOn thread is :" + Thread.currentThread().getName());
                    }
                }).subscribe(consumer);
    }

    /**
     * 看似很完美, 但我们忽略了一点, 如果在请求的过程中Activity已经退出了, 这个时候如果回到主线程去更新UI,
     * 那么APP肯定就崩溃了, 怎么办呢,Disposable 它是个开关, 调用它的dispose()方法时就会切断水管,
     * 使得下游收不到事件, 既然收不到事件, 那么也就不会再去更新UI了.
     * 因此我们可以在Activity中将这个Disposable 保存起来, 当Activity退出时, 切断它即可.
     * <p>
     * 那如果有多个Disposable 该怎么办呢, RxJava中已经内置了一个容器CompositeDisposable,
     * 每当我们得到一个Disposable时就调用CompositeDisposable.add()将它添加到容器中, 在退出的时候,
     * 调用CompositeDisposable.clear() 即可切断所有的水管.
     */
    public void sample4(String username) {
        RetrofitManager.getInstance().create(Api.GithubSevice.class).getUser(username)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<GitUser>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                    }

                    @Override
                    public void onNext(GitUser value) {
                        Log.d("GitUser--onNext", "GitUser: " + value.getAvatar_url() + "  " + value.getBlog());
                    }

                    @Override
                    public void onError(Throwable e) {
                    }

                    @Override
                    public void onComplete() {
                    }
                });
    }

}
